package com.captjack.bigdata.hadoop.hbase.base;

import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * hbase操作基类
 *
 * @author Jack Sparrow
 * @version 1.0.0
 * @date 16/08/2018 15:16
 * package com.sinoiov.hypt.bigdata
 */
public abstract class AbstractHadoopDatabaseRepository {

    /**
     * 列簇和列的分隔符
     */
    private static final char COLUMN_FAMILY_SPILT_CHAR = ':';

    protected Map<String, byte[]> findByRowKeyAsBytes(HTable hTable, byte[] rowKey, HbaseColumnFamilyEntity... hbaseColumnFamilyEntities) throws Exception {
        return this.findByRowKeyAsBytes(hTable, rowKey, true, hbaseColumnFamilyEntities);
    }

    protected void putByRowKey(HTable hTable, byte[] rowKey, HbaseColumnFamilyEntity... hbaseColumnFamilyEntities) throws Exception {
        this.putByRowKey(hTable, rowKey, true, hbaseColumnFamilyEntities);
    }

    /**
     * 写入
     *
     * @param hTable                    表
     * @param rowKey                    行键
     * @param isClose                   是否关闭表
     * @param hbaseColumnFamilyEntities 实体
     * @throws Exception 异常
     */
    protected void putByRowKey(HTable hTable, byte[] rowKey, boolean isClose, HbaseColumnFamilyEntity... hbaseColumnFamilyEntities) throws Exception {
        try {
            hTable.put(formatHbaseColumnFamilyEntity(rowKey, hbaseColumnFamilyEntities));
        } finally {
            if (isClose) {
                hTable.close();
            }
        }
    }

    /**
     * @param hTable                    hTable对象
     * @param rowKey                    rowKey
     * @param isClose                   是否关闭表
     * @param hbaseColumnFamilyEntities hbase实体
     * @return map
     * @throws Exception 异常
     */
    protected Map<String, byte[]> findByRowKeyAsBytes(HTable hTable, byte[] rowKey, boolean isClose, HbaseColumnFamilyEntity... hbaseColumnFamilyEntities) throws Exception {
        try {
            // 返回查询结果
            return this.readResultAsBytes(hTable.get(new Get(rowKey)), hbaseColumnFamilyEntities);
        } finally {
            if (isClose) {
                hTable.close();
            }
        }
    }

    /**
     * @param resultScanner             查询结果集
     * @param hbaseColumnFamilyEntities 查询条件
     * @return list
     * @throws Exception 异常
     */
    protected final List<Map<String, String>> readResultScannerAsString(ResultScanner resultScanner, boolean isClose, HbaseColumnFamilyEntity... hbaseColumnFamilyEntities) throws Exception {
        try {
            List<Map<String, String>> result = new ArrayList<Map<String, String>>();
            Result queryResult;
            while ((queryResult = resultScanner.next()) != null) {
                result.add(readResultAsString(queryResult, hbaseColumnFamilyEntities));
            }
            return result;
        } finally {
            if (isClose) {
                resultScanner.close();
            }
        }
    }

    /**
     * @param resultScanner             查询结果集
     * @param hbaseColumnFamilyEntities 查询条件
     * @return list
     * @throws Exception 异常
     */
    protected final List<Map<String, byte[]>> readResultScannerAsBytes(ResultScanner resultScanner, boolean isClose, HbaseColumnFamilyEntity... hbaseColumnFamilyEntities) throws Exception {
        try {
            List<Map<String, byte[]>> result = new ArrayList<Map<String, byte[]>>();
            Result queryResult;
            while ((queryResult = resultScanner.next()) != null) {
                result.add(readResultAsBytes(queryResult, hbaseColumnFamilyEntities));
            }
            return result;
        } finally {
            if (isClose) {
                resultScanner.close();
            }
        }
    }

    /**
     * @param queryResult               查询结果result
     * @param hbaseColumnFamilyEntities 读取
     * @return map
     */
    protected final Map<String, String> readResultAsString(Result queryResult, HbaseColumnFamilyEntity... hbaseColumnFamilyEntities) {
        if (queryResult == null) {
            return null;
        }
        Map<String, String> result = new HashMap<String, String>(hbaseColumnFamilyEntities.length);
        for (HbaseColumnFamilyEntity hbaseColumnFamilyEntity : hbaseColumnFamilyEntities) {
            String columnFamilyName = hbaseColumnFamilyEntity.getColumnFamilyName();
            for (String columnName : hbaseColumnFamilyEntity.getColumns().keySet()) {
                result.put(columnFamilyName + COLUMN_FAMILY_SPILT_CHAR + columnName, new String(queryResult.getValue(Bytes.toBytes(columnFamilyName), Bytes.toBytes(columnName))));
            }
        }
        return result;
    }

    /**
     * @param queryResult               查询结果result
     * @param hbaseColumnFamilyEntities 读取
     * @return map
     */
    protected final Map<String, byte[]> readResultAsBytes(Result queryResult, HbaseColumnFamilyEntity... hbaseColumnFamilyEntities) {
        if (queryResult == null) {
            return null;
        }
        Map<String, byte[]> result = new HashMap<String, byte[]>(hbaseColumnFamilyEntities.length);
        for (HbaseColumnFamilyEntity hbaseColumnFamilyEntity : hbaseColumnFamilyEntities) {
            String columnFamilyName = hbaseColumnFamilyEntity.getColumnFamilyName();
            for (String columnName : hbaseColumnFamilyEntity.getColumns().keySet()) {
                result.put(columnFamilyName + COLUMN_FAMILY_SPILT_CHAR + columnName, queryResult.getValue(Bytes.toBytes(columnFamilyName), Bytes.toBytes(columnName)));
            }
        }
        return result;
    }

    /**
     * 转换为put对象
     *
     * @param hbaseColumnFamilyEntities hbase实体
     * @return put对象
     */
    protected final Put formatHbaseColumnFamilyEntity(byte[] rowKey, HbaseColumnFamilyEntity... hbaseColumnFamilyEntities) {
        Put put = new Put(rowKey);
        for (HbaseColumnFamilyEntity hbaseColumnFamilyEntity : hbaseColumnFamilyEntities) {
            String columnFamilyName = hbaseColumnFamilyEntity.getColumnFamilyName();
            for (Map.Entry<String, byte[]> stringEntry : hbaseColumnFamilyEntity.getColumns().entrySet()) {
                String columnName = stringEntry.getKey();
                put.add(Bytes.toBytes(columnFamilyName), Bytes.toBytes(columnName), stringEntry.getValue());
            }
        }
        return put;
    }

}
